/*
 * Copyright (c) 2019 The StreamX Project
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.streamxhub.streamx.flink.core.conf

import com.streamxhub.streamx.common.conf.ConfigConst
import com.streamxhub.streamx.common.conf.ConfigConst.{KEY_FLINK_DEPLOYMENT_OPTION_PREFIX, KEY_FLINK_DEPLOYMENT_PROPERTY_PREFIX}
import com.streamxhub.streamx.common.util.PropertiesUtils
import org.apache.commons.cli.{DefaultParser, Options}

import java.net.URLClassLoader
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.util.{Failure, Success, Try}

object ParameterCli {

  private[this] val propertyPrefix = KEY_FLINK_DEPLOYMENT_PROPERTY_PREFIX
  private[this] val optionPrefix = KEY_FLINK_DEPLOYMENT_OPTION_PREFIX
  private[this] val optionMain = s"$propertyPrefix$$internal.application.main"

  lazy val flinkOptions: Options = FlinkRunOption.allOptions

  lazy val parser = new DefaultParser

  def main(args: Array[String]): Unit = print(read(args))

  def read(args: Array[String]): String = {
    args(0) match {
      case "--vmopt" =>
        //解决jdk1.8+ 动态加载资源到classpath下问题
        ClassLoader.getSystemClassLoader match {
          case c if c.isInstanceOf[URLClassLoader] => ""
          case _ => "--add-opens java.base/jdk.internal.loader=ALL-UNNAMED --add-opens jdk.zipfs/jdk.nio.zipfs=ALL-UNNAMED"
        }
      case action =>
        val conf = args(1)
        val map = Try(if (conf.endsWith(".properties")) {
          PropertiesUtils.fromPropertiesFile(conf)
        } else {
          PropertiesUtils.fromYamlFile(conf)
        }) match {
          case Success(value) => value
          case _ => Map.empty[String, String]
        }
        val programArgs = args.drop(2)
        action match {
          case "--option" =>
            val option = getOption(map, programArgs)
            val buffer = new StringBuffer()
            Try {
              val line = parser.parse(flinkOptions, option, false)
              line.getOptions.foreach(x => {
                buffer.append(s" -${x.getOpt}")
                if (x.hasArg) {
                  buffer.append(s" ${x.getValue()}")
                }
              })
            } match {
              case Failure(exception) => exception.printStackTrace()
              case _ =>
            }
            map.getOrElse(optionMain, null) match {
              case null =>
              case mainClass => buffer.append(s" -c $mainClass")
            }
            buffer.toString.trim
          case "--property" =>
            val buffer = new StringBuffer()
            map
              .filter(x => x._1 != optionMain && x._1.startsWith(propertyPrefix) && x._2.nonEmpty)
              .foreach(x => buffer.append(s" -D${x._1.drop(propertyPrefix.length)}=${x._2}"))
            buffer.toString.trim
          case "--name" =>
            map.getOrElse(propertyPrefix.concat(ConfigConst.KEY_FLINK_APP_NAME), "").trim match {
              case appName if appName.nonEmpty => appName
              case _ => ""
            }
          //是否detached模式...
          case "--detached" =>
            val option = getOption(map, programArgs)
            val line = parser.parse(FlinkRunOption.allOptions, option, false)
            val detached = line.hasOption(FlinkRunOption.DETACHED_OPTION.getOpt) || line.hasOption(FlinkRunOption.DETACHED_OPTION.getLongOpt)
            val mode = if (detached) "Detached" else "Attach"
            mode
          case _ => null
        }
    }
  }

  def getOption(map: Map[String, String], args: Array[String]): Array[String] = {
    val optionMap = new mutable.HashMap[String, Any]()
    map.filter(_._1.startsWith(optionPrefix)).filter(_._2.nonEmpty).filter(x => {
      val key = x._1.drop(optionPrefix.length)
      //验证参数是否合法...
      flinkOptions.hasOption(key)
    }).foreach(x => {
      Try(x._2.toBoolean).getOrElse(x._2) match {
        case b if b.isInstanceOf[Boolean] => if (b.asInstanceOf[Boolean]) optionMap += s"-${x._1.drop(optionPrefix.length)}".trim -> true
        case v => optionMap += s"-${x._1.drop(optionPrefix.length)}".trim -> v
      }
    })
    //来自从命令行输入的参数,优先级比配置文件高,若存在则覆盖...
    args match {
      case Array() =>
      case array =>
        Try {
          val line = parser.parse(flinkOptions, array, false)
          line.getOptions.foreach(x => {
            if (x.hasArg) {
              optionMap += s"-${x.getLongOpt}".trim -> x.getValue()
            } else {
              optionMap += s"-${x.getLongOpt}".trim -> true
            }
          })
        } match {
          case Failure(e) => e.printStackTrace()
          case _ =>
        }
    }
    val array = new ArrayBuffer[String]
    optionMap.foreach(x => {
      array += x._1
      if (x._2.isInstanceOf[String]) {
        array += x._2.toString
      }
    })
    array.toArray
  }

}
